# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: xguo

import collections
import enum
import json
import math
import typing

import tornado.gen

import coin.flow.subscriber as flow_subscriber
import coin.flow.topic_map as topic_map
import coin.proto.coin_order_enums_pb2 as coin_order_pb2

from coin.base.config import Config
from coin.base.timestamp import get_timestamp
from coin.exchange.base.order_fill_util import FillManager, LastFill
from coin.exchange.base.rest.task_scheduler import Scheduler, wrap_async_task
from coin.exchange.bithumb.kr_rest.constants import RETURN_CODE_SUCCESS, min_order_qty
from coin.exchange.bithumb.kr_rest.private_client import BithumbPrivateParser, to_float
from coin.exchange.bithumb.kr_rest.product import BithumbProduct, BithumbCurrency
from coin.exchange.util.rate_limit import RateLimiter
from coin.exchange.util.rest_rate_limit import RestRateLimiter
from coin.proto.coin_order_gateway_pb2 import OrderGatewayConfig, OrderEvent
from coin.exchange.bithumb.kr_rest.native_private_client import (
    BithumbNativePrivateClient,
    BithumbOrderSide,
    TransactionType,
)
from coin.exchange.base.rest_order_gateway_base import (
    RestOrderGateway,
    exchange_message_handler,
    PrivateExchangeMessage,
    create_order_internal,
    Order,
    OrderSide,
    OrderType,
    BaseBalanceOrderGatewayInfo,
    ProductFillElement,
    to_proto_order_side,
)
from coin.proto.coin_query_pb2 import (
    AccountBalance,)
'''
Transaction example
{
  "search": "2",
  "transfer_date": "1547452123810153",
  "units": "- 0.8",
  "price": 3199840,
  "btc1krw": 4001000,
  "fee": "960.24",
  "btc_remain": "1.01860131",
  "krw_remain": 111641618
}
'''
Transaction = collections.namedtuple('Transaction',
                                     [
                                         'search',
                                         'order_currency',
                                         'payment_currency',
                                         'transfer_date',
                                         'units',
                                         'unit_price',
                                         'fee',
                                         'fee_currency',
                                         'base_remain',
                                         'quote_remain',
                                     ])


def _transaction_type_to_order_side(tran_type: TransactionType):
  if tran_type == TransactionType.COMPLETED_PURCHASE:
    return OrderSide.BUY
  elif tran_type == TransactionType.COMPLETED_SALES:
    return OrderSide.SELL
  else:
    raise ValueError('Wrong transaction type! %s', tran_type)


def _is_transaction_buy_or_sell(transaction):
  buy_or_sell = (TransactionType.COMPLETED_PURCHASE, TransactionType.COMPLETED_SALES)
  return TransactionType(transaction.search) in buy_or_sell


def _is_no_working_orders_msg(msg):
  return msg == {'status': '5600', 'message': '거래 진행중인 내역이 존재하지 않습니다.'}


def _is_cancel_not_found_msg(msg):
  msg_list = [
      {
          'status': '5600', 'message': '매수건의 상태가 진행중이 아닙니다. 취소할 수 없습니다.'
      },
      {
          'status': '5600', 'message': '매도건의 상태가 진행중이 아닙니다. 취소할 수 없습니다.'
      },
      {
          'status': '5600', 'message': '대기건이 존재하지 않습니다.'
      },
  ]
  return msg in msg_list


def _is_empty_pending_list(msg):
  return msg == {'status': '5600', 'message': '대기주문이 없습니다.'}


def _to_bithumb_order_side(order_side: OrderSide):
  return getattr(BithumbOrderSide, order_side.name)


def _last_fill_from_transaction(transaction, timestamp):
  assert transaction.search in (TransactionType.COMPLETED_PURCHASE,
                                TransactionType.COMPLETED_SALES), transaction

  price_last_fill = to_float(transaction.unit_price)
  assert price_last_fill > 0, transaction
  qty_last_fill = abs(to_float(transaction.units))

  last_fill = LastFill(timestamp_last_fill=timestamp,
                       price_last_fill=price_last_fill,
                       qty_last_fill=qty_last_fill)
  return last_fill


def _to_transaction_record(transaction_msg, order_currency, payment_currency):
  base_remain = transaction_msg.pop('order_balance')
  quote_remain = transaction_msg.pop('payment_balance')
  unit_price = transaction_msg.pop('price')
  fee = transaction_msg.pop('fee')
  fee_currency = transaction_msg.pop('fee_currency')
  transfer_date = int(transaction_msg.pop('transfer_date')) * 1000

  transaction_dict = dict()
  transaction_dict['order_currency'] = order_currency
  transaction_dict['payment_currency'] = payment_currency
  transaction_dict['base_remain'] = base_remain
  transaction_dict['quote_remain'] = quote_remain
  transaction_dict['fee'] = fee
  transaction_dict['fee_currency'] = fee_currency
  transaction_dict['unit_price'] = unit_price
  transaction_dict['transfer_date'] = transfer_date
  transaction_dict['units'] = transaction_msg['units'].replace(' ', '')
  transaction_dict['search'] = int(transaction_msg['search'])
  transaction = Transaction(**transaction_dict)
  return transaction


def create_order_from_proto(order_id, proto_order):
  side_map = {
      coin_order_pb2.BUY_ORDER: OrderSide.BUY,
      coin_order_pb2.SELL_ORDER: OrderSide.SELL,
  }
  price_orig = proto_order.price
  qty_orig = proto_order.qty
  order_side = side_map[proto_order.side]
  order_type = OrderType.LIMIT
  exchange_order_id = proto_order.order_id
  product = BithumbProduct.FromStrProduct(proto_order.symbol)

  fill = FillManager(
      qty_orig=qty_orig,
      price_orig=price_orig,
      qty_accum_fill=0.0,
      price_avg_fill=None,
      qty_last_fill=None,
      price_last_fill=None,
  )

  order = Order(order_id=order_id,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=order_side,
                order_type=order_type,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=create_order_internal(
                    timestamp=get_timestamp(),
                    exchange_order_id=exchange_order_id,
                    fill=fill,
                ))
  return order


def create_order_from_transaction(transaction, timestamp):
  last_fill = _last_fill_from_transaction(transaction, timestamp)
  price = last_fill.price_last_fill
  qty = last_fill.qty_last_fill
  fee = to_float(transaction.fee)
  product = BithumbProduct.FromStrProduct(
      '%s-%s' % (transaction.order_currency.currency, transaction.payment_currency.currency))
  fill = FillManager(
      qty_orig=qty,
      price_orig=price,
      qty_accum_fill=qty,
      price_avg_fill=price,
      qty_last_fill=qty,
      price_last_fill=price,
      timestamp_last_fill=timestamp,
  )
  order_side = _transaction_type_to_order_side(transaction.search)
  order = Order(order_id='ledger',
                product=product,
                price=price,
                qty=qty,
                order_side=order_side,
                order_type=OrderType.LIMIT,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=create_order_internal(
                    timestamp=timestamp,
                    exchange_order_id=None,
                    fill=fill,
                    fee=fee,
                ))
  return order


class BithumbOrderGatewayInfo(BaseBalanceOrderGatewayInfo):
  CurrencyType = BithumbCurrency
  ProductType = BithumbProduct

  def __init__(
      self,
      currencies: typing.List[BithumbCurrency],
      products: typing.List[BithumbProduct],
  ):
    super().__init__(currencies=currencies, products=products)
    self._num_balance_updates = 0
    self._balance_changed = collections.defaultdict(bool)
    self._quote = BithumbCurrency.FromStr('KRW')

  def update_balance(self, update_msg):
    account_balance = BithumbPrivateParser.parse_balance(update_msg)
    self._num_balance_updates += 1
    for balance in account_balance.each_balance:
      currency = BithumbCurrency.FromStrCurrency(balance.currency)
      if currency in self.currencies:
        if self._num_balance_updates > 2:
          total = self.get_balance(currency)
          self._balance_changed[currency] = not math.isclose(total, balance.total, abs_tol=1e-5)
        self.set_balance(currency, balance.total, balance)

  def balance_as_proto(self):
    return AccountBalance(exchange='Bithumb',
                          market_type='Spot',
                          each_balance=self.get_notNone_balances_internal())


# The rate limit message can be  {"status": "5600", "message": "Please try again."}
# or {"status": "5600", "message": "Please try again"},
# or {"status": "5600", "message": "Please try again!"},
# sometime this message comes with status_code = 429, sometimes not.
def is_bithumb_rate_limit_msg(message):
  return message['status'] == '5600' and 'Please try again' in message['message']


def is_no_transaction_history_msg(message):
  no_trans_msg = {'message': '거래 체결내역이 존재하지 않습니다.', 'status': '5600'}
  return message == no_trans_msg


class BithumbOrderGateway(RestOrderGateway):
  _exchange = 'Bithumb'
  _market_type = 'Spot'
  _api_version = None

  OGInfoType = BithumbOrderGatewayInfo

  class _MsgType(enum.IntEnum):
    ACCOUNT_BALANCE = 0
    OPEN_ORDERS = 1
    ORDER_DETAILS = 2
    FIRST_TRANSACTION = 3
    TRANSACTION_INFO = 4

  def __init__(self,
               currencies: typing.List[BithumbCurrency],
               products: typing.List[BithumbProduct],
               config: Config,
               og_config=None,
               logger=None):
    super().__init__(currencies, products, config, og_config, logger)

    self._private_client = None
    self._scheduler = None
    self._ioloop = None
    self._flow_sub = None
    self._publisher = None

    default_og_config = OrderGatewayConfig(
        account_query_period=2,
        orders_snapshot_query_period=5,
        fill_query_period=3,
        hit_rate_limit_cooldown_time=12,
        burst_query_avoid_time=0.3,
        order_rate_limit=1,
        order_rate_limit_per=4,
        unacked_order_timeout=30,
        unacked_cancel_timeout=5,
        use_ledger=True,
        use_fill_checker=True,
    )
    default_og_config.MergeFrom(self._og_config)
    self._og_config = default_og_config

    # Handle fill message
    self._processed_fill_ids = set()
    self._rest_rate_limiter = RestRateLimiter(delay=5)
    self._order_rate_limiter = RateLimiter(
        self._og_config.order_rate_limit,
        self._og_config.order_rate_limit_per,
    )
    self._user_transactions = collections.defaultdict(set)
    self._latest_transfer_date = collections.defaultdict(int)
    self._maybe_filled = {}

  def start(self, queue, ioloop, flow_sub=None):
    api_host = self._og_config.rest_root_url \
               if self._og_config.HasField('rest_root_url') else None
    self._private_client = BithumbNativePrivateClient(key_file=self._config.key_file,
                                                      api_host=api_host,
                                                      use_async=True)
    topic_id = topic_map.search_topic_id('bithumb_rest_private_1')[0][1]
    self._ioloop = ioloop
    self._publisher = queue.get_writer(topic_id=topic_id)
    self._scheduler = Scheduler(ioloop=ioloop, logger=self._logger)
    self._install_periodic_query()

    # Subscribe to flow for feed.
    self._flow_sub = flow_sub or flow_subscriber.from_queue(queue)
    self._flow_sub.subscribe('bithumb_rest_private_1', self.on_private_exchange_msg)
    self._logger.info(
        'order_rate_limit: %s, order_rate_limit_per: %s',
        self._og_config.order_rate_limit,
        self._og_config.order_rate_limit_per,
    )

  def submittable(self):
    return (self.is_ready() and self._rest_rate_limiter.is_available()
            and self._order_rate_limiter.check())

  @tornado.gen.coroutine
  def _install_periodic_query(self):
    assert self._scheduler is not None, 'self._scheduler is not set!'
    assert self._publisher is not None, 'self._publisher is not set!'

    # query balance
    query_balance_task = wrap_async_task(
        self._private_client.query_info_balance,
        msg_type=self._MsgType.ACCOUNT_BALANCE,
        publisher=self._publisher,
    )
    self._scheduler.add_periodic_task(period=self._og_config.account_query_period,
                                      job_func=query_balance_task)

    for product in self._og_info.products:
      yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)

      # query active orders
      query_orders_task = wrap_async_task(self._private_client.query_info_orders,
                                          msg_type=self._MsgType.OPEN_ORDERS,
                                          publisher=self._publisher,
                                          func_params=dict(
                                              order_currency=product.base.native_currency,
                                              payment_currency=product.quote.native_currency))
      self._scheduler.add_periodic_task(period=self._og_config.orders_snapshot_query_period,
                                        job_func=query_orders_task)

      # query transactions
      query_transaction_task = wrap_async_task(self._private_client.query_info_user_transactions,
                                               msg_type=self._MsgType.FIRST_TRANSACTION,
                                               publisher=self._publisher,
                                               func_params=dict(
                                                   order_currency=product.base.native_currency,
                                                   payment_currency=product.quote.native_currency))
      self._scheduler.run_once(job_func=query_transaction_task)

    def check_maybe_filled():
      timestamp = get_timestamp()
      for order in list(self._maybe_filled.values()):
        delta_time = (timestamp - order.internal.timestamp) / 1e9
        if delta_time > 600:
          self._maybe_filled.pop(order.internal.exchange_order_id)
      self._logger.info('MAYBE_FILLED dict size = %s', len(self._maybe_filled))

    self._scheduler.add_periodic_task(period=10, job_func=check_maybe_filled)
    self.start_scheduler()

  @tornado.gen.coroutine
  def _on_hit_order_rate_limit(self):
    if not self.is_ready():
      return

    self._og_ready.set_reinit()
    self._scheduler.stop()
    self._logger.info('Hit rate limit, disable OG for % seconds.',
                      self._og_config.hit_rate_limit_cooldown_time)
    yield tornado.gen.sleep(self._og_config.hit_rate_limit_cooldown_time)
    self._scheduler.start()
    self._logger.info('Bithumb OG is active again.')

  def _async_submit_order(self, order, post_only=False):
    if post_only:
      self._logger.warning('post only not supported.')
    assert order.order_id is not None, order
    assert order.order_type == OrderType.LIMIT, order

    min_qty = min_order_qty.get(order.product.base.currency, 0.)
    if order.qty < min_qty:
      self._logger.error('order qty is less than min order qty! %s, %s', order.product, order.qty)
      self._order_manager.remove_order(order)
      return

    formatter = self._order_formatter[order.product]
    params = {
        'order_currency': order.product.base.native_currency,
        'payment_currency': order.product.quote.native_currency,
        'price': formatter.format_price(order.price),
        'qty': formatter.format_qty(order.qty),
        'order_side': _to_bithumb_order_side(order.order_side),
    }
    fut = self._private_client.query_trade_place(**params)
    return fut

  def _async_cancel_order(self, order):
    assert order.internal.exchange_order_id is not None, order
    fut = self._private_client.query_trade_cancel(
        order_currency=order.product.base.native_currency,
        payment_currency=order.product.quote.native_currency,
        order_id=order.internal.exchange_order_id,
        order_side=_to_bithumb_order_side(order.order_side))
    return fut

  def _handle_submit_response(self, order, response):
    """ {
    "status"    : "0000",
    "order_id"  : "1428646963419",
    "data": [
        {
            "cont_id"   : "15313",
            "units"     : "0.61460000",
            "price"     : "284000",
            "total"     : 174546,
            "fee"       : "0.00061460"
        },
        {
            "cont_id"   : "15314",
            "units"     : "0.18540000",
            "price"     : "289000",
            "total"     : 53581,
            "fee"       : "0.00018540"
        }
    ]}
    {
      'status': '0000', 'order_id': '1547806145099262',
      'data': [{
        'contNo': '25638838',
        'units': '126.5889',
        'price': 2757,
        'total': 349006,
        'fee': '0.03797667'}]}
    """
    msg = self._preprocess_response(response)
    if not msg:
      self._order_manager.remove_order(order, ignore_error=True)
      return OrderEvent.ORDER_REJECTED

    if msg.get('status') != RETURN_CODE_SUCCESS:
      self._order_manager.remove_order(order)
      if is_bithumb_rate_limit_msg(msg):
        self._logger.error('Rate limit hit (%s, %s), slow down!\n%s',
                           response.status_code,
                           order.order_id,
                           response.content)
        self._rest_rate_limiter.set_error(status_code=response.status_code)
      elif msg == {'status': '5600', 'message': '매수금액이 사용가능 KRW 를 초과하였습니다.'}:
        self._logger.error('Insufficient balance!')
      elif msg['status'] == '5600' and '최소' in msg['message']:
        self._logger.error('Order quantity is smaller than min order qty! %s\n%s', order, msg)
      else:
        self._logger.error('error msg: %s', msg)
      return OrderEvent.ORDER_ERROR

    if 'order_id' not in msg:
      self._logger.error('Order submission error! %s', msg)
      self._order_manager.remove_order(order)
      return OrderEvent.ORDER_ERROR

    exchange_order_id = str(msg['order_id'])
    if msg.get('data'):
      self._logger.info('Order filled immediately!, %s', msg)

    prev_order = self._order_manager.get(exchange_order_id=exchange_order_id)
    if prev_order is None:
      self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
    else:
      self._logger.info('order id replaced: %s => %s', prev_order.order_id, order.order_id)
      self._order_manager.remove_order(order)
      self._order_manager.remove_order(prev_order)
      prev_order.order_id = order.order_id
      self._order_manager.add_order(prev_order)
    return OrderEvent.ORDER_ACCEPTED

  def _handle_cancel_response(self, order, response):
    """ Cancel response has a very simple form:
    {"status" : "0000"}
    """
    msg = self._preprocess_response(response)
    exchange_order_id = order.internal.exchange_order_id
    if not msg:
      order.cancel_sent = False
      return OrderEvent.CANCEL_ERROR

    if msg['status'] == RETURN_CODE_SUCCESS:
      if self._order_manager.is_dead_order(order):
        # Order status maybe set already by other message handler.
        self._logger.info('cancel-successful order is dead already! %s', order)
      else:
        self._order_manager.set_order(order, cancel_confirmed=True)
      self._maybe_filled.pop(exchange_order_id, None)
      return OrderEvent.CANCEL_CONFIRMED
    elif _is_cancel_not_found_msg(msg):
      self._logger.info('Order maybe filled already! %s', order)
      if not self._order_manager.is_dead_order(order):
        self._order_manager.set_order(order, cancel_confirmed=True)
      self._maybe_filled[exchange_order_id] = order
      return OrderEvent.CANCEL_CONFIRMED
    else:
      order.cancel_sent = False
      self._logger.error('Cancel order error! %s\n%s', order, msg)
      return OrderEvent.CANCEL_ERROR

  @exchange_message_handler(_MsgType.ACCOUNT_BALANCE)
  def _handle_account_balance(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    update_msg = self._preprocess_response(response)
    if not update_msg:
      return

    self._og_info.update_balance(update_msg)
    self._og_logger.write_balance(self._og_info.balance_as_proto())

    if not self.is_ready():
      self._og_ready.set_balance_ready(True)
      self._og_ready.check_ready()
      self._logger.info('Balance is ready!')

  @exchange_message_handler(_MsgType.OPEN_ORDERS)
  def _handle_open_orders_info(self, exchange_message: PrivateExchangeMessage):
    timestamp = exchange_message.received_timestamp
    response = exchange_message.data
    update_msg = self._preprocess_response(response)
    if not update_msg:
      return

    request_params = exchange_message.request_params
    order_currency = BithumbCurrency.FromStrNativeCurrency(request_params['order_currency'])
    payment_currency = BithumbCurrency.FromStrNativeCurrency(request_params['payment_currency'])
    product = BithumbProduct.FromStrProduct('%s-%s' %
                                            (order_currency.currency, payment_currency.currency))
    if product not in self._og_info.products:
      return

    if update_msg['status'] == RETURN_CODE_SUCCESS:
      proto_order_list = list(BithumbPrivateParser.parse_orders(update_msg, product).each_order)
    elif _is_no_working_orders_msg(update_msg):
      proto_order_list = list()
    else:
      self._logger.error('Invalid response message! %s', update_msg)
      return

    query_exchange_order_ids = set(
        proto_order.exchange_order_id for proto_order in proto_order_list)
    exchange_order_ids = set()

    # Handle orders that appear in OG's working order list but not in exchange's active order list.
    for order in self.get_working_order():
      if order.product != product:  # Skip
        continue

      if order.accepted is False:  # Skip
        continue

      exchange_order_id = order.internal.exchange_order_id
      exchange_order_ids.add(exchange_order_id)
      if exchange_order_id not in query_exchange_order_ids:
        if timestamp - order.internal.timestamp < 5e9:
          # If order is just added, it may not appear immediately in the active order list.
          continue

        # order not in active order list, possible reasons:
        #   1. Order is successfully canceled by OG (cancel_sent flag is set), but cancel response hasn't been received yet. # noqa: E501
        #   2. Order is successfully canceled by another program (cancel_sent flat is not set)
        #   3. Order is filled.
        if order.cancel_sent:
          self._logger.info(
              'Order canceled, but cancel response not received yet, order can be filled or canceled.\n%s',  # noqa: E501
              order)
          # Remove it from working orders, treat it as cancel confirmed. Does not hurt if later we found it is actually fully filled. # noqa: E501
          self._order_manager.set_order(order, cancel_confirmed=True)
        else:
          # Since the order is disappeared from actively order list, it is filled or canceled by another program. # noqa: E501
          self._logger.info(
              'Order not in active order list, probably filled or canceled by another program.\n%s',
              order)
          # Remove it from working orders treat it as fully filled. Does not hurt if later we found it is actually canceled by another program. # noqa: E501
          self._order_manager.set_order(order, fully_filled=True)
        self._maybe_filled[exchange_order_id] = order

    # Handle orders that appear in exchange active order list, but not in OG's working order list
    for proto_order in proto_order_list:
      if proto_order.order_id not in exchange_order_ids:
        self._logger.info('Active order not found in OG working order list, exchange order id %s',
                          proto_order.order_id)
        order = self._order_manager.get(exchange_order_id=proto_order.exchange_order_id)
        if order is not None:
          if order.internal.cancel_confirmed and timestamp - order.internal.cancel_sent_time < 10e9:
            # If order is just cancel confirmed, it may not be removed immediately from active order list. # noqa: E501
            continue
          else:
            # Rarely happen.
            self._logger.error('order state error: %s\nproto_order: %s', order, proto_order)
            self._order_manager.remove_order(order)
            order.cancel_sent = False
            self._order_manager.add_order(order)
        else:
          order_id = self._order_id_generator.gen_order_id(is_foreign=True)
          order = create_order_from_proto(order_id, proto_order)
          self._order_manager.add_order(order)
          self._logger.info('Add foreign order: %s', order)

    if not self.is_ready():
      product_list = getattr(self._og_ready, 'open_orders_product_list', set())
      product_list.add(product)
      self._og_ready.open_orders_product_list = product_list
      self._logger.info('first query open_order product list %s',
                        json.dumps(list(product_list), indent=2, default=str))
      if product_list == set(self._og_info.products):
        self._og_ready.set_orders_ready(True)
        self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.FIRST_TRANSACTION)
  def _handle_first_transactions(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    update_msg = self._preprocess_response(response)
    if not update_msg:
      return

    native_order_currency = exchange_message.request_params['order_currency']
    native_payment_currency = exchange_message.request_params['payment_currency']

    order_currency = BithumbCurrency.FromStrNativeCurrency(native_order_currency)
    payment_currency = BithumbCurrency.FromStrNativeCurrency(native_payment_currency)
    product = BithumbProduct.FromStrProduct('%s-%s' %
                                            (order_currency.currency, payment_currency.currency))

    for transaction_msg in update_msg['data']:
      transaction = _to_transaction_record(transaction_msg, order_currency, payment_currency)
      if not _is_transaction_buy_or_sell(transaction):
        continue  # Ignore on buy/sell transaction

      self._user_transactions[product].add(transaction)
      if transaction.transfer_date > self._latest_transfer_date[product]:
        self._latest_transfer_date[product] = transaction.transfer_date

    query_transaction_task = wrap_async_task(self._private_client.query_info_user_transactions,
                                             msg_type=self._MsgType.TRANSACTION_INFO,
                                             publisher=self._publisher,
                                             func_params=dict(
                                                 order_currency=native_order_currency,
                                                 payment_currency=native_payment_currency))
    self._scheduler.add_periodic_task(period=self._og_config.fill_query_period,
                                      job_func=query_transaction_task)

  @exchange_message_handler(_MsgType.TRANSACTION_INFO)
  def _handle_transactions(self, exchange_message: PrivateExchangeMessage):
    timestamp = exchange_message.received_timestamp
    response = exchange_message.data
    update_msg = self._preprocess_response(response)
    if not update_msg:
      return

    native_order_currency = exchange_message.request_params['order_currency']
    native_payment_currency = exchange_message.request_params['payment_currency']

    order_currency = BithumbCurrency.FromStrNativeCurrency(native_order_currency)
    payment_currency = BithumbCurrency.FromStrNativeCurrency(native_payment_currency)
    product = BithumbProduct.FromStrProduct('%s-%s' %
                                            (order_currency.currency, payment_currency.currency))

    for transaction_msg in update_msg['data']:
      transaction = _to_transaction_record(transaction_msg, order_currency, payment_currency)
      if not _is_transaction_buy_or_sell(transaction):
        continue  # Skip

      if transaction not in self._user_transactions[product]:
        self._user_transactions[product].add(transaction)
        if transaction.transfer_date > self._latest_transfer_date[product]:
          self._latest_transfer_date[product] = transaction.transfer_date
        else:
          self._logger.error('Reversed transfer_date %s, %s',
                             transaction,
                             self._latest_transfer_date[product])

        if self._og_config.use_ledger:
          last_fill = _last_fill_from_transaction(transaction, timestamp)
          order = create_order_from_transaction(transaction, timestamp)
          fill_type = self._guess_fill_type(order,
                                            last_fill.price_last_fill,
                                            transaction.transfer_date)
          fee = abs(to_float(transaction.fee))
          fee_currency = transaction.fee_currency
          if transaction.search not in (TransactionType.COMPLETED_PURCHASE,
                                        TransactionType.COMPLETED_SALES):
            ValueError('Invalid transaction type here! %s', transaction.search)

          fill_proto = ProductFillElement(
              fill_id='no_fill_id',
              price=last_fill.price_last_fill,
              qty=last_fill.qty_last_fill,
              order_id=order.order_id,
              side=to_proto_order_side(order.order_side),
              guess_fill_type=fill_type,
              symbol=order.product.symbol,
              fee=fee,
              fee_currency=fee_currency,
          )
          self.process_fill(fill_proto, timestamp, order)

  def _preprocess_response(self, response):
    if response.status_code >= 500:
      self._logger.error(
          '5XX server timeout error! %s, %s, %s',
          response.status_code,
          response.content,
          response.url,
      )
      return None
    elif response.status_code == 429:
      self._ioloop.add_callback(self._on_hit_order_rate_limit)
      self._rest_rate_limiter.set_error(response.status_code)
      self._logger.error(
          'Rate limit hit (429)! %s, %s, %s',
          response.status_code,
          response.content,
          response.url,
      )
      return None
    elif response.status_code != 200:
      self._logger.error('response error! %s, %s, %s',
                         response.status_code,
                         response.content,
                         response.url)
      return None

    update_msg = response.json()
    if is_bithumb_rate_limit_msg(update_msg):
      self._logger.error('Rate limit: %s', update_msg)
      self._ioloop.add_callback(self._on_hit_order_rate_limit)
      return None

    if update_msg['status'] != RETURN_CODE_SUCCESS:
      if _is_no_working_orders_msg(update_msg):
        pass
      elif _is_cancel_not_found_msg(update_msg):
        pass
      elif _is_empty_pending_list(update_msg):
        return None
      else:
        self._logger.error(
            'Rest query error! %s\n%s\n%s\n%s',
            update_msg,
            response.status_code,
            response.content,
            response.url,
        )
        return None

    return update_msg

  def _guess_fill_type(self, order, fill_price, fill_timestamp):
    fill_type = coin_order_pb2.TAKER_FILL_TYPE
    for recent_order in self._maybe_filled.values():
      if recent_order.product != order.product:
        continue

      if math.isclose(recent_order.price, fill_price, abs_tol=1e-4):
        fill_type = coin_order_pb2.MAKER_FILL_TYPE
        break
    return fill_type
